package com.tensai.study.新概念.响应式编程.reactor;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.time.Duration;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

@Slf4j
public class FluxSinkTest {

	public static void main(String[] args) throws InterruptedException {
		Flux.interval(Duration.ofSeconds(1), Duration.ofSeconds(1)).subscribe(System.out::println);
		Thread.sleep(5000);
	}

	@Test
	public void test() throws InterruptedException {
		final Flux<Integer> flux = Flux.create(fluxSink -> {
			//NOTE sink:class reactor.core.publisher.FluxCreate$SerializedSink
			log.info("sink:{}", fluxSink.getClass());
			while (true) {
				log.info("sink next");
				fluxSink.next(ThreadLocalRandom.current().nextInt());
			}
		}, FluxSink.OverflowStrategy.BUFFER);

		//NOTE flux:class reactor.core.publisher.FluxCreate,prefetch:-1
		log.info("flux:{},prefetch:{}", flux.getClass(), flux.getPrefetch());

		flux.subscribe(e -> {
			log.info("subscribe:{}", e);

			try {
				TimeUnit.SECONDS.sleep(10);
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
		});

		TimeUnit.MINUTES.sleep(20);
	}
}
